#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "database.h"

#include "MQTTClient.h"
#include "mqtt_request_handle.h"


#include "mqtt_client.h"
#include "log_print.h"

#include "common.h"
#include "ll.h"
#include "cJSON.h"

extern database_json_t db_config;


/*
    内存泄漏测试命令：valgrind --tool=memcheck --leak-check=yes --show-reachable=yes ./XhDataCenter

    有关cjson内存释放：
        1、在cjson中释放内存时，只需要释放父节点就可以，不需要专门管理子节点，子节点会随着父节点的释放一并释放掉
        2、cJSON_Print会单独申请内存，需要调用cJSON_free释放内存
*/


enum {
    MQTT_CLIENT_CREAT,      /* 创建client */
    MQTT_CLIENT_SET_CB,     /* client设置callback */
    MQTT_CLIENT_CONNECT,    /* client连接到broker */
    MQTT_CLIENT_LICENSE_1,  /* 校验license步骤1 */
    MQTT_CLIENT_LICENSE_2,  /* 校验license步骤2 */
    MQTT_CLIENT_SUBSCRIBE,  /* 订阅主题 */
    MQTT_CLIENT_RUN,        /* client正常运行 */
    MQTT_CLIENT_DESTROY     /* 删除client */
};


#define QOS             1
#define TIMEOUT         10000L
#define BROKER_ADDRESS  "tcp://localhost:1883"
#define CLIENT_ID       "XhDataCenterMQTTClient"

static MQTTClient client;
static int mqtt_state = MQTT_CLIENT_CREAT;  /* mqtt状态机 */
volatile MQTTClient_deliveryToken deliveredtoken;


struct topic_table {
    char *request;          /* 订阅的主题 */
    char *request_keyword;  /* 订阅的主题字符里面的关键字，主要是去掉通配符"+"和"#" */
    char *response;         /* 回复的主题 */
    int (*fun)(char* app, char *tpc, char *msg, char *response);   /* 处理函数 */
};


struct topic_table topic_table[] = {

    /* 查询版本信息 */
    {"+/get/request/datacenter/version",            "/get/request/datacenter/version",              "datacenter/get/response/%s/version",           topic_get_versin},
    /* 模型名称查询 */
    {"+/get/request/datacenter/modelschema",        "/get/request/datacenter/modelschema",          "datacenter/get/response/%s/modelschema",       topic_get_modelschema},
    /* 指定模型删除 */
    {"+/action/request/datacenter/deletemodel",     "/action/request/datacenter/deletemodel",       "datacenter/action/response/%s/deletemodel",    topic_action_deletemodel},
    /* 模型内容查询 */
    {"+/get/request/datacenter/model",              "/get/request/datacenter/model",                "datacenter/get/response/%s/model",             topic_get_model},
    /* 模型设置 */
    {"+/set/request/datacenter/model",              "/set/request/datacenter/model",                "datacenter/set/response/%s/model",             topic_set_model},

    /* 设备 GUID 查询 */
    {"+/get/request/datacenter/guid",               "/get/request/datacenter/guid",                 "datacenter/get/response/%s/guid",              topic_get_guid},
    /* 设备/APP 注册 */
    {"+/set/request/datacenter/register",           "/set/request/datacenter/register",             "datacenter/set/response/%s/register",          topic_set_register},
    // {"+/set/request/datacenter/register",           "/set/request/datacenter/register",             "datacenter/set/response/%s/register",          topic_set_register_V2},
    /* 设备/APP 注册关系查询 */
    {"+/get/request/datacenter/register",           "/get/request/datacenter/register",             "datacenter/get/response/%s/register",          topic_get_register},
    /* 设备/APP 取消注册 */
    {"+/action/request/datacenter/unregister",      "/action/request/datacenter/unregister",        "datacenter/action/response/%s/unregister",     topic_action_unregister},

    /* 实时数据与冻结数据写 */
    {"+/notify/event/datacenter/+/+",               "/notify/event/datacenter/",                    "datacenter/notify/response/%s/%s/%s",          topic_notify_event},
    /* 按设备读取实时数据 */
    {"+/get/request/datacenter/realtime",           "/get/request/datacenter/realtime",             "datacenter/get/response/%s/realtime",          topic_get_realtime},

    /* 获取子设备状态 */
    {"+/get/request/datacenter/status",             "/get/request/datacenter/status",               "datacenter/get/response/%s/status",            topic_get_dev_status},

    /* 监听广播子设备在离线状态 */
    {"+/notify/spont/*/+/+",                        "/notify/spont/",                               "datacenter/notify/response/%s/%s/%s",          topic_set_dev_status}

};


struct mqtt_rx_s {
    char *topic;
    char *msg;
};


static ll_t *mqtt_rx_list;                      /* list，存储mqtt请求 */
static pthread_rwlock_t mqtt_rx_list_rwlock;    /* 定义list的读写锁 */


/**
  * @brief  mqtt_rx_list初始化
  * @param  null
  * @retval null
  */
int mqtt_rx_list_init(void)
{
    int res;

    log_printf(MSG_DEBUG, "mqtt_rx_list_init");

    mqtt_rx_list = ll_new(ll_no_teardown);

    res = pthread_rwlock_init(&mqtt_rx_list_rwlock, NULL);  //初始化读写锁
    if (res != 0) {
        LOG_PRINTF_ERROR( "mqtt_rx_list_rwlock initialization failed");
        exit(EXIT_FAILURE);
    }

    return 0;
}


/**
  * @brief  动态申请mqtt_rx_s内存
  * @param  null
  * @retval null
  */
struct mqtt_rx_s * creat_mqtt_rx_buff(int topic_len, int msg_len)
{
    struct mqtt_rx_s *p;

    if ((topic_len <= 0) || (msg_len <= 0)) return 0;

    p = (struct mqtt_rx_s *)malloc(sizeof(struct mqtt_rx_s));

    if (p) {
        memset(p, 0x00, sizeof(struct mqtt_rx_s));
    } else {
        LOG_PRINTF_ERROR( "func(%s) malloc error", __FUNCTION__);
        return 0;
    }

    topic_len = topic_len + 1;  /* 字符串多一个字节 */
    msg_len = msg_len + 1;      /* 字符串多一个字节 */
    p->topic = (char *)malloc(topic_len);
    p->msg = (char *)malloc(msg_len);

    if ((p->topic) && (p->msg)) {
        memset(p->topic, 0x00, topic_len);
        memset(p->msg, 0x00, msg_len);
    } else {
        if (p->topic) {
            free(p->topic);
            p->topic = NULL;
        }
        if (p->msg) {
            free(p->msg);
            p->msg = NULL;
        }
        if (p) {
            free(p);
            p = NULL;
        }
        LOG_PRINTF_ERROR( "func(%s) error", __FUNCTION__);
        return 0;
    }

    return p;
}


/**
  * @brief  创建list结点，存储mqtt请求
  * @param  topic mqtt topic
  * @param  topic mqtt message
  * @retval null
  */
int creat_mqtt_rx_node(char *topic, char *msg)
{
#define TOPIC_LEN_MAX 128
#define MSG_LEN_MAX   8192

    struct mqtt_rx_s *p = NULL;

    if ((NULL == topic) || (NULL == msg)) return -1;

    if ((strlen(topic) > TOPIC_LEN_MAX) || (strlen(msg) > MSG_LEN_MAX)) {
        LOG_PRINTF_ERROR( "func(%s) mqtt request oversize(topic: %d bytes, payload: %d bytes):\n topic[%s] \n msg[%s]", __FUNCTION__, strlen(topic), strlen(msg), topic, msg);
        return -2;
    }

    p = creat_mqtt_rx_buff(strlen(topic), strlen(msg));

    if (NULL == p) return -3;

    memcpy(p->topic, topic, strlen(topic));
    memcpy(p->msg, msg, strlen(msg));

    pthread_rwlock_wrlock(&mqtt_rx_list_rwlock);
    ll_insert_last(mqtt_rx_list, p);
    pthread_rwlock_unlock(&mqtt_rx_list_rwlock);

    return 0;
}


/**
  * @brief  清除mqtt请求的list
  * @param  null
  * @retval null
  */
int mqtt_rx_list_destroy(void)
{
    int i;
    struct mqtt_rx_s *p;
    ll_t *list = mqtt_rx_list;

    log_printf(MSG_DEBUG, "mqtt_rx_list_destroy");

    if (NULL == list) return -1;

    pthread_rwlock_wrlock(&mqtt_rx_list_rwlock);
    for (i = 0; i < list->len; i++) {
        p = (struct mqtt_rx_s *)ll_get_n(list, i);
        if (p->topic) {
            free(p->topic);
            p->topic = NULL;
        }
        if (p->msg) {
            free(p->msg);
            p->msg = NULL;
        }
        if (p) {
            free(p);
            p = NULL;
        }
    }

    ll_delete(mqtt_rx_list);
    mqtt_rx_list = NULL;
    pthread_rwlock_unlock(&mqtt_rx_list_rwlock);
    pthread_rwlock_destroy(&mqtt_rx_list_rwlock);

    return 0;
}

#if 0
cJSON * jn_mqtt_request;

/**
  * @brief  mqtt_rx_list初始化
  * @param  null
  * @retval null
  */
int jn_mqtt_request_init(void)
{
    log_printf(MSG_DEBUG, "jn_mqtt_request_init");

    jn_mqtt_request = cJSON_CreateObject();

    return 0;
}
#endif

/**
  * @brief  发布消息
  * @param  topic 消息主题
  * @param  payload 消息内容
  * @param  payload_len 消息内容字节数
  * @retval null
  */
int mqtt_publish_message(char *topic, char *payload, int payload_len)
{
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    if ((NULL == topic) || (NULL == payload)) return -1;

    pubmsg.payload = payload;
    pubmsg.payloadlen = payload_len;
    pubmsg.qos = QOS;
    pubmsg.retained = 0;

    log_printf(MSG_INFO, "Message publish, topic:%s", topic);
    log_printf(MSG_VERBOSE, "Message publish, len:%d, payload: %s", payload_len, payload);
    // log_printf(MSG_DEBUG, "Message publish: topic: %s \n, len:%d, payload: %s \n\n", topic, payload_len, payload);

    if ((rc = MQTTClient_publishMessage(client, topic, &pubmsg, &token)) != MQTTCLIENT_SUCCESS) {
        LOG_PRINTF_ERROR( "Failed to publish message: topic[%s] payload[%s], return code %d", topic, payload, rc);
        return -2;
    }

    return 0;
}


/**
  * @brief  从topic获取app名字
  * @param  null
  * @retval null
  */
static int get_app_name(char *request, char *name)
{
    int i;
    int num;
    char *p = NULL;

    if ((NULL == request) || (NULL == name)) return -1;

    //printf("%s %s \n", __FUNCTION__, request);

    p = strstr(request, "/");
    if (NULL == p) return -2;

    //printf("request = %d, p = %d", (int)request, (int)p);
    num = p - request;

    for (i = 0; i < num; i++) {
        name[i] = request[i];
    }

    return 0;
}


/**
  * @brief  mqtt有消息到达
  * @param  null
  * @retval null
  */
static int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    // int i;
    int err = 0;

    log_printf(MSG_INFO, "Message arrived, topic:%s", topicName);
    log_printf(MSG_VERBOSE, "Message arrived: payload: %s", message->payload);

    /* 收到mqtt请求先不着急处理，缓存起来集中处理，这样可以提高效率 */
    err = creat_mqtt_rx_node(topicName, message->payload);
    if (0 != err) {
        LOG_PRINTF_ERROR( "func(creat_mqtt_rx_node) error[%d]", err);
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);

    return 1;
}


/**
  * @brief  mqtt消息发生成功回调函数
  * @param  null
  * @retval null
  */
static void delivered(void *context, MQTTClient_deliveryToken dt)
{
#if 0   /* 测试 */
    static int count = 0;
    printf("count = %d \n", ++count);
#endif

    //printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}


/**
  * @brief  mqtt客户端与broker连接断开的时候会调用这个函数
  * @param  null
  * @retval null
  */
static void connlost(void *context, char *cause)
{
    LOG_PRINTF_ERROR( "func(%s) MQTT Connection lost, cause: %s", __FUNCTION__, cause);
    mqtt_state = MQTT_CLIENT_DESTROY;   /* 复位状态机，重新连接broker */
}


/**
  * @brief  mqtt订阅主题
  * @param  p_client mqtt客户端
  * @param  topic 要订阅的主题
  * @param  qos
  * @retval null
  */
static int subscribe_topic(MQTTClient *p_client, char *topic, int qos)
{
    int rc;

    if ((rc = MQTTClient_subscribe(*p_client, topic, qos)) != MQTTCLIENT_SUCCESS) {
        LOG_PRINTF_ERROR( "subscribe topic[%s] err", topic);
        rc = EXIT_FAILURE;
    } else {
        log_printf(MSG_INFO, "subscribe topic[%s] ok, qos[%d]", topic, qos);
    }

    return rc;
}


/**
  * @brief  mqtt正常运行，处理mqtt请求
  * @param  null
  * @retval null
  */
static void mqtt_client_run(void)
{
    int i;
    int j;
    ll_t *list = mqtt_rx_list;
    struct mqtt_rx_s *p;
    int len;
    int rtn;

    //log_printf(MSG_VERBOSE, "MQTT client run");

    if (NULL == list) return;

    len = list->len;
    for (j = 0; j < len; j++) {   /* 从列表获取mqtt请求并处理 */
        p = (struct mqtt_rx_s *)ll_get_n(list, j);
        if (p) {
            log_printf(MSG_INFO, "Message handle: topic: %s \npayload: %s", p->topic, (char*)p->msg);

            for (i = 0; i < (sizeof(topic_table) / sizeof(struct topic_table)); i++) {
                if (NULL != strstr(p->topic, topic_table[i].request_keyword)) {
                    char app_name[128];
                    memset(app_name, 0x00, sizeof(app_name));
                    if (get_app_name(p->topic, app_name)) continue;

                    log_printf(MSG_INFO, "%s request topic: %s", app_name, topic_table[i].request);
                    if (NULL != topic_table[i].fun) {
                        rtn = topic_table[i].fun(app_name, p->topic, p->msg, topic_table[i].response);
                        if (rtn) {
                            /* 解析出错，打印错误的mqtt请求包 */
                            log_printf(MSG_ERROR ,"Message error[%d]: topic: %s \npayload: %s ", rtn, p->topic, (char*)p->msg);
                        }
                    }
                    else
                    {
                        LOG_PRINTF_ERROR( "Message error: topic: %s \npayload: %s", p->topic, (char *)p->msg);
                    }
                    break;
                }
                // else
                // {
                //     LOG_PRINTF_ERROR( "Message error: topic: %s \npayload: %s", p->topic, (char*)p->msg);
                // }
            }
        }
    }

    pthread_rwlock_wrlock(&mqtt_rx_list_rwlock);
    /* 请求都处理完成，清空list里面的node，但是保留list */
    while(len) {
        p = (struct mqtt_rx_s *)ll_get_n(mqtt_rx_list, 0);
        if (p) {
            if (p->topic) {
                free(p->topic);
                p->topic = NULL;
            }
            if (p->msg) {
                free(p->msg);
                p->msg = NULL;
            }
            if (p) {
                free(p);
                p = NULL;
            }
        }
        ll_remove_n(mqtt_rx_list, 0);
        len--;
    }
    pthread_rwlock_unlock(&mqtt_rx_list_rwlock);
}


/**
  * @brief  mqtt状态机
  * @param  null
  * @retval null
  */
static void mqtt_client_to_broker(void)
{
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;
    int i;

    switch (mqtt_state) {
        case MQTT_CLIENT_CREAT:
            if (0 == strlen(db_config.mqtt_broker)) {
                break;
            }
            memset(&client, 0x00, sizeof(client));
            mqtt_rx_list_init();
            if ((rc = MQTTClient_create(&client, db_config.mqtt_broker, CLIENT_ID, MQTTCLIENT_PERSISTENCE_NONE, NULL)) != MQTTCLIENT_SUCCESS) {
                LOG_PRINTF_ERROR( "MQTT Failed to create client, return code %d", rc);
                sleep(2);
            } else {
                mqtt_state = MQTT_CLIENT_SET_CB;
            }
            break;

        case MQTT_CLIENT_SET_CB:
            if ((rc = MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered)) != MQTTCLIENT_SUCCESS) {
                LOG_PRINTF_ERROR( "MQTT Failed to set callbacks, return code %d", rc);
                sleep(2);
                mqtt_state = MQTT_CLIENT_DESTROY;
            } else {
                mqtt_state = MQTT_CLIENT_CONNECT;
            }
            break;

        case MQTT_CLIENT_CONNECT:
            conn_opts.keepAliveInterval = 20;
            conn_opts.cleansession = 1;

            if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
                LOG_PRINTF_ERROR( "MQTT Failed to connect, return code %d", rc);
                sleep(2);
                mqtt_state = MQTT_CLIENT_DESTROY;
            } else {
                mqtt_state = MQTT_CLIENT_SUBSCRIBE;
            }
            break;

        case MQTT_CLIENT_SUBSCRIBE:
            for (i = 0; i < (sizeof(topic_table) / sizeof(struct topic_table)); i++) {
                if (NULL != topic_table[i].request) {
                    subscribe_topic(&client, topic_table[i].request, QOS);
                }
            }
            mqtt_state = MQTT_CLIENT_RUN;
            break;

        case MQTT_CLIENT_RUN:
            if (MQTTClient_isConnected(client)) {
                /* 正常 */
                mqtt_client_run();
                //usleep(2);
            } else {
                /* 与broker的连接断开了，需要重连 */
                LOG_PRINTF_ERROR( "client disconnect");
                mqtt_state = MQTT_CLIENT_DESTROY;
            }
            break;

        case MQTT_CLIENT_DESTROY:
            mqtt_rx_list_destroy();
            MQTTClient_destroy(&client);
            mqtt_state = MQTT_CLIENT_CREAT;
            break;

        default:
            mqtt_state = MQTT_CLIENT_DESTROY;
            break;
    }
}


/**
  * @brief  mqtt线程
  * @param  null
  * @retval null
  */
static void *mqtt_client_thread(void *parg)
{
    log_printf(MSG_INFO, "%s create success.", __FUNCTION__);

    while (is_app_running()) {
        mqtt_client_to_broker();
        usleep(1000);
    }

    return NULL;
}


/**
  * @brief  mqtt初始化
  * @param  null
  * @retval null
  */
void mqtt_client_init(void)
{
    pthread_t tid;
    int err;

    /* 创建线程 */
    err = pthread_create(&tid, NULL, mqtt_client_thread, NULL);

    if (-1 == err) {
        LOG_PRINTF_ERROR( "mqtt_client_thread create fail.");
    }
}


/**
  * @brief  mqtt客户端销毁
  * @param  null
  * @retval null
  */
void mqtt_client_destory(void)
{
    log_printf(MSG_INFO, "end mqtt_client destroy.");
    MQTTClient_destroy(&client);
}

